Friday Code and Notes (Week 4)

class CounterValue {
    /* 1. Volatile in C, and volatile in Java, are completely different

       In C: volatile is a way to tell the compiler that
             reads and writes to this variable
             shouldn't be optimised away.

       In Java: makes sure reads and writes to this variable are
                sequentially consistent.

       Reads/writes are *sequentially consistent* if the way they
       behave are consistent with atomic reads and writes.
       In other words: the concurrency model we've used
         has assumed that as soon as I write,
         the results become immediately observable to everybody.

       |----------------|-------------|
       | int x,y = 0;                 |
       |----------------|-------------|
       | x := 1         | y := 1      |
       | int z := y     | int a := x  |
       --------------------------------

      Q: What are the possible final values of (a,z) now?
       Can we get (1,1)?
       - Yes, if they both execute in lockstep.
       How about (0,1) or (1,0)?
       - Yes, if one process finishes before the other starts
       How about (0,0)?
       - With atomic writes, no.
       - Under *weak memory models*, yes.
         W/ low-level concurrent programming on modern multicore
          archs, you sometimes can observe results
          that don't arise from any interleaving of our reads
          and writes.
     */    
    private volatile int count;

    public CounterValue() {
        count = 0;
    }

    public void increment() {
        count++;
    }

    public int read() {
        return count;
    }
}

class CounterThread extends Thread {

    private CounterValue c;

    public CounterThread(CounterValue c) {
        this.c = c;
    }

    public void run() {
        for(int i = 0; i < 5000; i++) {
            // here, responsibility for synching access to the resource
            // belongs to the threads
            // or in other words: responsibility is diffused throughout
            // possibly the whole program

            // wait on semaphore here
            c.increment();
            // signals on a semaphore here
        }
    }
}

public class Counter {

    public static void main(String [] main) throws InterruptedException {
        CounterValue c = new CounterValue();
        Thread t1 = new CounterThread(c);
        Thread t2 = new CounterThread(c);

        t1.start();
        t2.start();

        t1.join(); // wait until t1 is finished
        t2.join();
        System.out.println("Final counter value is " + c.read());
    }

}

class CounterValue {
    private volatile int count;

    public CounterValue() {
        count = 0;
    }

    /* A *synchronized* method
        can only have one process at a time in it.
       I.e., it's implicitly protected by a lock.
       For each object, the synchronized methods
       use the same lock.
     */
    public synchronized void increment() {
        count++;
    }

    // In this case, synching on read is pointless
    public synchronized int read() {
        return count;
    }
}

class CounterThread extends Thread {

    private CounterValue c;

    public CounterThread(CounterValue c) {
        this.c = c;
    }

    public void run() {
        for(int i = 0; i < 5000; i++) {
            c.increment();
        }
    }
}

public class Counter2 {

    public static void main(String [] main) throws InterruptedException {
        CounterValue c = new CounterValue();
        Thread t1 = new CounterThread(c);
        Thread t2 = new CounterThread(c);

        t1.start();
        t2.start();

        t1.join(); // wait until t1 is finished
        t2.join();
        System.out.println("Final counter value is " + c.read());
    }

}

/* A monitor for managing the buffer */
/* A fake buffer: we're only going to track
   whether it's full or not, and not put 
   anything in it.
 */
class Buffer {
    private int max_size;
    private volatile int current_size;

    public Buffer(int max_size) {
        this.max_size = max_size;
        this.current_size = 0;
    }

    public synchronized void enqueue() {
        /* The queue might be full.
           If so, we wait.
         */
        while(max_size == current_size) {
            // wait is "wait on a condition variable"
            // every Object in java implicitly has
            // exactly one condition variable.
            try {
                wait();
            }
            catch(InterruptedException e) {}
            // when we're here, we don't know if
            // the condition we were waiting
            // is *still* true
        }
        /* here, we really do know that max_size != current_size */
        current_size++;
        notifyAll();        
    }

    public synchronized void dequeue() {
        while(current_size == 0) {
            try {
                wait();
            }
            catch(InterruptedException e) {}
        }
        current_size--;
        notifyAll();        
    }
}

class Consumer extends Thread {
    Buffer b;

    public Consumer(Buffer b) {
        this.b = b;
    }

    public void run() {
        while(true) {
            b.enqueue();
            System.out.println("Thread #" + getId() + " enqueued a thing!");
        }
    }
}

class Producer extends Thread {
    Buffer b;

    public Producer(Buffer b) {
        this.b = b;
    }

    public void run() {
        while(true) {
            b.dequeue();
            System.out.println("Thread #" + getId() + " dequeued a thing!");
        }
    }
}

public class ProdCom {
    public static void main(String[] args) {
        Buffer b = new Buffer(1);
        Thread t1 = new Producer(b);
        Thread t2 = new Producer(b);
        Thread t3 = new Consumer(b);
        Thread t4 = new Consumer(b);
        t1.start();
        t2.start();
        t3.start();
        t4.start();
    }
}